package cn.tedu.sql

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.types.Row

import java.util.UUID
import scala.util.Random

/**
 * @author Amos
 * @date 2022/5/24
 */

object StreamSqlDemo {
  def main(args: Array[String]): Unit = {
    // 1. 构建流处理环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 2. 流处理表环境的构建
    val tableEnv = StreamTableEnvironment.create(env)
    // 3.指定以eventTime处理数据
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 创建自定义数据源
    /*
    创建一个订单样例类Order ，包含四个字段（订单ID、用户ID、订单金额、时间戳）
    a. 使用for 循环生成1000 个订单
    b. 随机生成订单ID（UUID）
    c. 随机生成用户ID（0-2）
    d. 随机生成订单金额（0-100）
    e. 时间戳为当前系统时间
    f. 每隔1 秒生成一个订单
     */
    import org.apache.flink.api.scala._
    val source: DataStream[Order1] = env.addSource(new SourceFunction[Order1] {

      var isRunnig = true

      // 生成数据并输出
      override def run(ctx: SourceFunction.SourceContext[Order1]): Unit = {
        for (i <- 0 to 1000 if isRunnig) {
          // 订单ID、用户ID、订单金额、时间戳
          val order = Order1(
            UUID.randomUUID().toString,
            Random.nextInt(3),
            Random.nextInt(101),
            System.currentTimeMillis())

          Thread.sleep(1000)

          // 收集并返回
          ctx.collect(order)
        }
      }

      // 取消生成的数据
      override def cancel(): Unit = {
        isRunnig = false
      }
    })

    // 指定watermark
    val waterMarkStream: DataStream[Order1] = source.assignTimestampsAndWatermarks(
      // 指定watermark的时间和抽取时间戳的字段
      new BoundedOutOfOrdernessTimestampExtractor[Order1](Time.seconds(2)) {
        override def extractTimestamp(element: Order1): Long = {
          element.createTime
        }
      }
    )

    // 注册成一张表  订单ID、用户ID、订单金额、时间戳
    import org.apache.flink.table.api.scala._
    tableEnv.createTemporaryView(
      "t_order",
      waterMarkStream,
      'orderid,'userid,'money,'createTime.rowtime)

    // sql  使用Flink SQL 来统计5 秒内用户的订单总数、订单的最大金额、订单的最小金额
    val sql =
      """
        |select
        |userid,
        |count(1) totalCount,
        |max(money) MaxMoney,
        |min(money) MinMoney
        |from t_order
        |group by
        |tumble(createTime,interval '5' second),
        |userid
        |
        |""".stripMargin

    // 执行sql
    val table: Table = tableEnv.sqlQuery(sql)

    // 得到结果，打印输出
    val result: DataStream[(Boolean, Row)] = tableEnv.toRetractStream[Row](table)
    result.print()
    env.execute()

  }

}

case class Order1(orderid: String, userid: Int, money: Int, createTime: Long)
